package com.leeyu.kafka.selfserial;

import org.apache.kafka.common.errors.SerializationException;
import org.apache.kafka.common.serialization.Serializer;

import java.nio.ByteBuffer;
import java.util.Map;

public class UserSerializer implements Serializer<User> {

    public void configure(Map<String, ?> configs, boolean isKey) {

    }

    @Override
    public byte[] serialize(String topic, User user) {
        try {
            if (user == null) {
                return null;
            }
            byte[] name;
            int nameSize;

            if (user.getName() != null) {
                name = user.getName().getBytes();
                nameSize = name.length;
            } else  {
                name = new byte[0];
                nameSize = 0;
            }
            /*
            id的长度4个字节，字符串的长度描述4个字节
            字符串本身长度nameSize个字节
             */

            ByteBuffer buffer = ByteBuffer.allocate(4 + 4 +nameSize);
            buffer.putInt(user.getId());
            buffer.putInt(nameSize);
            buffer.put(name);
            return buffer.array();
        } catch (Exception e) {
            throw new SerializationException("serialize User error", e);
        }
    }

}
